package com.fourth.utils;

import com.fourth.bean.TransientSink;
import com.fourth.common.GmallConfig;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author wjy
 * @create 2022-08-20 9:35
 */
public class MyClickHouseUtil {
    //因为是工具类 所以输入类型不能写具体的javaBean，可以定义成泛型 但是要在返回值的前面加上泛型方法，表示T是泛型
    public static <T> SinkFunction<T> getSink(String sql){
        return JdbcSink.<T>sink(sql,
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        //思考：需要取出字段,给sql占位符赋值 -> 需要遍历t 泛型怎么遍历？-> 遍历javaBean ->反射暴力获取字段
                        Class<?> clz = t.getClass();

                        //反射怎么使用? 反过来调用
                        //例：获取属性：obj.getXXX() -> XXX.get(obj) 调用方法：obj.xxx(p1,p2,p3) ->xxx.invoke(obj,p1,p2,p3)
                        //反射:invoke(T t,Object...) 第二个参数是可变形参，因为不知道方法有几个参数

                        //获取字段
                        Field[] fields = clz.getDeclaredFields();

                        int j = 0;
                        for (int i = 0; i < fields.length; i++) {
                            Field field = fields[i];

                            //禁用安全检查
                            field.setAccessible(true);

                            //如果有需要忽略写出的字段，并且加了注解，那么需要处理一下 首先获取注解
                            TransientSink transientSink = field.getAnnotation(TransientSink.class);
                            if (transientSink != null){
                                j++;
                                continue;
                            }

                            Object value = field.get(t);
                            preparedStatement.setObject(i + 1 - j, value);
                        }
                    }
                }, JdbcExecutionOptions.builder()
                        .withBatchIntervalMs(1000L)  //达到1s就写一次 这些参数可以作为形参从外部传进来，更加灵活，这里数据量比较小，就直接给了
                        .withBatchSize(5)  //达到5条就写一次
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()
        );
    }
}
